package com.allen.flink.batch.sql;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

/**
 * 功能:表转流
 *
 * @date: 2020-03-25 15:51
 * @author: Allen
 * @version: 0.0.4-snapshot
 * @Email: allenZyhang@163.com
 * @since: JDK 1.8
 **/
public class TableToStream {
    public static void main(String[] args) throws Exception {
        //获取运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        //创建TableEnvironment
        StreamTableEnvironment streamTabEnv = StreamTableEnvironment.create(env);
        
        //读取数据源  读取网络数据
        DataStream<String> ds = env.socketTextStream("127.0.0.1", 9999, "\n");
        
        //解析数据
        DataStream<Tuple2<String, Integer>> dsMap = ds.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(final String s, final Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] arr = s.toLowerCase().split("\\W+");//非单词
                for (String word : arr) {
                    if (word.length() > 0) {
                        out.collect(new Tuple2<>(word, 1));
                    }
                }
            }
        });
        
        //流转表
        Table table = streamTabEnv.fromDataStream(dsMap, "word,frequency");
        streamTabEnv.registerTable("wc", table);
        
        streamTabEnv.sqlQuery("select word, count(frequency) from wc group by word");
        
        //Append 模式 表转流
        // DataStream<Row> ds2 = streamTabEnv.toAppendStream(table, Row.class);
        
        //Retract 模式
        DataStream<Tuple2<Boolean, Row>> ds2 = streamTabEnv.toRetractStream(table, Row.class);
        
        ds2.print();
        
        env.execute("Flink Table To Stream");
    }
}
